package com.mango.iot.gateway.coap.server.resource;

import com.mango.iot.gateway.coap.common.CoapResourceConst;
import com.mango.iot.gateway.coap.mq.KafkaProducer;
import com.mango.iot.gateway.coap.mq.msg.CoapExchangeMessage;
import com.mango.iot.gateway.coap.utils.TopicMatcherUtil;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.californium.core.Utils;
import org.eclipse.californium.core.coap.CoAP;
import org.eclipse.californium.core.server.resources.CoapExchange;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.Map;

/**
 * <pre>
 * 主题数据上报
 * </pre>
 *
 * @author liangfeihu
 * @since 2023/6/28 16:09
 */
@Slf4j
@Component
public class TopicResource extends BaseCoapResource {

    @Autowired
    private KafkaProducer kafkaProducer;

    public TopicResource() {
        super(CoapResourceConst.TOPIC);
        getAttributes().setTitle("Topic Resource");
    }

    @Override
    public void handlePOST(CoapExchange exchange) {
        log.info("[TopicResource] request={}", Utils.prettyPrint(exchange.advanced().getRequest()));

        String topic = getParamValueFromUriQuery(exchange, CoapResourceConst.TOPIC);
        if (StringUtils.isEmpty(topic)) {
            exchange.respond(CoAP.ResponseCode.BAD_REQUEST);
            return;
        }
        TopicMatcherUtil matcherUtil = new TopicMatcherUtil(CoapResourceConst.EVENT_TOPIC);
        if (!matcherUtil.matches(topic)) {
            exchange.respond(CoAP.ResponseCode.UNSUPPORTED_CONTENT_FORMAT);
            return;
        }

        Map<String, String> topicMap = parseTopic(topic);
        String token = getToken(exchange);
        if (!validateToken(topicMap.get("productKey"), topicMap.get("deviceNo"), token)) {
            exchange.respond(CoAP.ResponseCode.UNAUTHORIZED);
            return;
        }

        //if(!tenantHandler.judgeDeviceStatusAvailable(topicMap.get("productKey"), topicMap.get("deviceNo"))){
        if (!true) {
            exchange.respond(CoAP.ResponseCode.FORBIDDEN);
            return;
        }
        this.process(topic, exchange.getRequestText());

        exchange.respond(CoAP.ResponseCode.CONTENT);
    }

    /**
     * <pre>
     * 处理业务消息1
     * </pre>
     *
     * @param topic
     * @param payload
     * @return void
     */
    private void process(String topic, String payload) {
        log.info("topic={}, payload={}", topic, payload);
        CoapExchangeMessage exchangeMessage = new CoapExchangeMessage();
        exchangeMessage.setTopic(topic);
        exchangeMessage.setPayload(payload);
        kafkaProducer.sendMessage("iot_coap_event", exchangeMessage);
    }

}
